package com.weds.data_sync.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.weds.data_sync.annotation.TableProcessorAnnotation;
import com.weds.data_sync.constant.ScriptServiceConstant;
import com.weds.data_sync.mapper.DataSyncRecordHistoryMapper;
import com.weds.data_sync.mapper.DataSyncRecordMapper;
import com.weds.data_sync.pojo.entity.DataSyncRecordDO;
import com.weds.data_sync.pojo.entity.DataSyncRecordHistoryDO;
import com.weds.data_sync.processor.TableProcessor;
import com.weds.data_sync.service.RecordHistoryService;
import com.weds.data_sync.service.RecordService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.stream.Collectors;

/**
 * <p>
 * 数据同步记录表 服务实现类
 * </p>
 *
 * @author lhw
 * @since 2023-12-29
 */
@Slf4j
@Service
public class RecordServiceImp extends ServiceImpl<DataSyncRecordMapper, DataSyncRecordDO> implements RecordService {

    @Autowired
    private RecordHistoryService recordHistoryService;

    @Autowired
    private DataSyncRecordMapper recordMapper;

    @Autowired
    private List<TableProcessor> processorList;

    @Autowired
    private DataSyncRecordHistoryMapper recordHistoryMapper;

    @Transactional
    @Override
    synchronized public void analysePlanRecord() {
        DataSyncRecordHistoryDO previousAnalyseInfo = recordHistoryService.getPreviousAnalyseInfo();
        List<DataSyncRecordDO> recordDOList = recordMapper.selectList(new QueryWrapper<DataSyncRecordDO>()
                .gt("id", Long.parseLong(previousAnalyseInfo.getEndRecordId()))
                .orderByAsc("id"));
        if (recordDOList.isEmpty()) {
            log.info("没有要处理的同步数据");
            return;
        }

        Map<String, List<DataSyncRecordDO>> groupRecordMap = recordDOList.stream()
                .collect(Collectors.groupingBy(DataSyncRecordDO::getTableName));

        if (Objects.isNull(processorList) || processorList.isEmpty()) {
            log.info("没有表类型处理器");
            throw new RuntimeException("没有表类型处理器");
        }

        groupRecordMap.forEach((recordKey, recordValue) -> {
            Optional<TableProcessor> processorOptional = processorList.stream().filter(i -> {
                TableProcessorAnnotation annotation = i.getClass().getAnnotation(TableProcessorAnnotation.class);
                String tableName = annotation.value();
                return recordKey.equalsIgnoreCase(tableName);
            }).findAny();
            if (processorOptional.isPresent()) {
                TableProcessor tableProcessor = processorOptional.get();
                tableProcessor.send(recordValue);
            } else {
                log.error("未找到合适的类型处理器。table: {}, data: {}", recordKey, recordValue);
                throw new RuntimeException("未找到合适的类型处理器");
            }
        });

        String startId = recordDOList.get(0).getId().toString();
        String endId = recordDOList.get(recordDOList.size() - 1).getId().toString();

        DataSyncRecordHistoryDO recordHistoryDO = DataSyncRecordHistoryDO.builder()
                .startRecordId(startId)
                .endRecordId(endId)
                .isSuccess(ScriptServiceConstant.CommonStatus.SUCCESS)
                .ctDate(LocalDateTime.now())
                .ltDate(LocalDateTime.now())
                .build();

        recordHistoryMapper.insert(recordHistoryDO);
    }
}
